package http

import (
	"bufio"
	"bytes"
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"strings"
	"sync/atomic"
	"time"

	"tunnel_manager/core"

	"github.com/quic-go/quic-go"
	"golang.org/x/sync/semaphore"
)

// ========================================
// HTTP协议处理器实现
// 实现 core.ProtocolHandler 接口
// ========================================

// HTTP2TunnelClient HTTP/2客户端接口
type HTTP2TunnelClient interface {
	SendResponseZeroCopy(requestID string, responseReader io.Reader) error
	IsConnected() bool
	GetActiveStreams() int64
}

// HTTP2ClientPool HTTP/2客户端连接池接口（避免循环依赖）
type HTTP2ClientPool interface {
	SelectConnection() HTTP2TunnelClient
	HasConnectedClient() bool
}

// Handler HTTP协议处理器
type Handler struct {
	tunnelID   string
	targetHost string
	targetPort string
	httpClient *http.Client

	// 🚀 HTTP/2混合模式：优先使用HTTP/2发送响应
	http2Pool HTTP2ClientPool

	// 并发控制
	concurrencySem *semaphore.Weighted
	maxConcurrency int64
	activeTasks    int64

	// 统计信息
	stats core.ProtocolStats
}

// contextReader 实现可取消的Reader
type contextReader struct {
	reader io.Reader
	ctx    context.Context
}

func (cr *contextReader) Read(p []byte) (n int, err error) {
	type result struct {
		n   int
		err error
	}

	ch := make(chan result, 1)
	go func() {
		n, err := cr.reader.Read(p)
		ch <- result{n: n, err: err}
	}()

	select {
	case res := <-ch:
		return res.n, res.err
	case <-cr.ctx.Done():
		return 0, cr.ctx.Err()
	}
}

// NewHandler 创建HTTP协议处理器
func NewHandler(tunnelID, targetHost, targetPort string) *Handler {
	// 🔥 提升并发限制：从200提升到1000，支持高并发场景
	maxConcurrency := int64(1000)

	// 🔥🔥🔥 关键优化：自定义Dialer，强制使用IPv4，避免IPv6连接失败
	dialer := &net.Dialer{
		Timeout:   10 * time.Second,
		KeepAlive: 30 * time.Second,
		// 🔥 禁用IPv6回退，直接使用IPv4
		FallbackDelay: -1, // 禁用Happy Eyeballs算法
	}

	transport := &http.Transport{
		// 🔥🔥🔥 强制使用IPv4拨号
		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
			// 将"tcp"强制改为"tcp4"，只使用IPv4
			if network == "tcp" {
				network = "tcp4"
			}
			return dialer.DialContext(ctx, network, addr)
		},
		
		// 🔥🔥🔥 连接池优化：支持1000+并发（关键性能提升）
		// 问题：1000并发时请求耗时8-12秒，连接数不足导致排队
		// 优化：大幅增加连接池大小，消除连接等待时间
		MaxIdleConns:        2000,              // 🚀 2倍提升：全局空闲连接池
		MaxIdleConnsPerHost: 2000,              // 🚀 4倍提升：每个主机的空闲连接
		MaxConnsPerHost:     2000,              // 🚀 设置最大连接数（原来是0=无限制，改为2000避免资源耗尽）
		IdleConnTimeout:     90 * time.Second,  // 🔥 恢复空闲超时为90秒（避免僵尸连接）

		// 🔥 性能优化：启用Keep-Alive和压缩
		DisableKeepAlives:  false,
		DisableCompression: false,
		ForceAttemptHTTP2:  false,

		// 🔥🔥🔥 超时配置优化：缩短超时以提升并发响应速度
		// 原因：1000并发时部分请求耗时12秒，需要快速释放失败连接
		TLSHandshakeTimeout:   5 * time.Second,  // TLS握手超时
		ResponseHeaderTimeout: 60 * time.Second, // 🚀 从120秒降至60秒（快速失败，避免长时间占用连接）
		ExpectContinueTimeout: 1 * time.Second,  // 100-continue超时

		// 🚀🚀🚀 缓冲区优化：增大缓冲区以支持高并发小请求
		// 分析：1000并发时大多是小请求（274 bytes），减少系统调用次数更重要
		MaxResponseHeaderBytes: 0,           // 不限制响应头大小
		WriteBufferSize:        128 * 1024,  // 🔥 降至128KB（小请求更高效）
		ReadBufferSize:         128 * 1024,  // 🔥 降至128KB（减少内存占用）
	}

	handler := &Handler{
		tunnelID:       tunnelID,
		targetHost:     targetHost,
		targetPort:     targetPort,
		maxConcurrency: maxConcurrency,
		concurrencySem: semaphore.NewWeighted(maxConcurrency),
		activeTasks:    0,
		httpClient: &http.Client{
			Transport: transport,
			// 🚀 优化：不设置总超时，避免大文件传输中断
			// 改由ResponseHeaderTimeout(20s)控制响应头，ReadTimeout自然终止
			Timeout: 0, // 不限制总超时（避免大文件60秒后中断）
		},
	}

	log.Printf("✅ [HTTP协议] 处理器已创建: Target=%s:%s, MaxConcurrency=%d, 并发等待超时=30s",
		targetHost, targetPort, maxConcurrency)

	return handler
}

// SetHTTP2Pool 设置HTTP/2连接池（用于混合模式）
func (h *Handler) SetHTTP2Pool(pool HTTP2ClientPool) {
	h.http2Pool = pool
	log.Printf("🚀 [HTTP协议] HTTP/2混合模式已启用")
}

// HandleRequest 实现 ProtocolHandler 接口
func (h *Handler) HandleRequest(stream quic.Stream, msg *core.TunnelMessage) error {
	requestID := msg.RequestID
	startTime := time.Now()
	log.Printf("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
	log.Printf("📥 [HTTP协议] 收到请求: RequestID=%s, Size=%d bytes",
		requestID[:8]+"...", len(msg.Body))
	fmt.Fprintf(os.Stderr, "⏰ [客户端] 收到请求时间: %s, RequestID=%s\n", time.Now().Format("15:04:05.000"), requestID[:8]+"...")

	// 并发控制
	// 🔥🔥🔥 关键修复：并发控制的超时不应影响HTTP请求
	// 使用独立的context仅用于semaphore获取，延长超时到30秒以支持高并发排队
	semCtx, semCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer semCancel()

	acquireStart := time.Now()
	if err := h.concurrencySem.Acquire(semCtx, 1); err != nil {
		log.Printf("❌ [HTTP协议] 并发超限: %v (等待时间=%.2fs, 当前活跃=%d, 最大并发=%d)", 
			err, time.Since(acquireStart).Seconds(), atomic.LoadInt64(&h.activeTasks), h.maxConcurrency)
		errorMsg := core.NewErrorMessage(h.tunnelID, requestID, 1, "并发超限")
		errorMsg.Write(stream)
		return err
	}
	defer h.concurrencySem.Release(1)
	acquireDuration := time.Since(acquireStart)

	// 🔥 性能监控：如果获取并发锁耗时>100ms，记录日志
	if acquireDuration > 100*time.Millisecond {
		log.Printf("⚠️  [性能] 并发锁等待时间: %.2fs, RequestID=%s", acquireDuration.Seconds(), requestID[:8]+"...")
	}

	atomic.AddInt64(&h.activeTasks, 1)
	defer atomic.AddInt64(&h.activeTasks, -1)
	atomic.AddInt64(&h.stats.TotalRequests, 1)

	// 解析HTTP请求
	parseStart := time.Now()
	req, err := http.ReadRequest(bufio.NewReader(bytes.NewReader(msg.Body)))
	parseDuration := time.Since(parseStart)
	if err != nil {
		log.Printf("❌ [HTTP协议] 解析请求失败: %v", err)
		atomic.AddInt64(&h.stats.ErrorCount, 1)
		errorMsg := core.NewErrorMessage(h.tunnelID, requestID, 2, fmt.Sprintf("解析失败: %v", err))
		errorMsg.Write(stream)
		return err
	}

	// 🔥 性能监控：如果解析耗时>50ms，记录日志
	if parseDuration > 50*time.Millisecond {
		log.Printf("⚠️  [性能] HTTP解析耗时: %.2fms, RequestID=%s", parseDuration.Seconds()*1000, requestID[:8]+"...")
	}

	// 修改目标URL
	targetURL := fmt.Sprintf("http://%s:%s%s", h.targetHost, h.targetPort, req.URL.Path)
	if req.URL.RawQuery != "" {
		targetURL += "?" + req.URL.RawQuery
	}

	// 🔥 打印服务端请求的目标地址
	log.Printf("🌐 [HTTP协议] 服务端请求地址: %s %s, RequestID=%s", 
		req.Method, targetURL, requestID[:8]+"...")

	// 🔥🔥🔥 关键优化:监听Stream关闭
	// 当服务端关闭Stream时（浏览器刷新/断开），立即停止处理
	streamCtx := stream.Context()

	// 创建新请求 - 直接使用streamCtx，无额外超时限制
	// streamCtx会在以下情况自动取消：
	// 1. 浏览器断开连接（刷新/关闭）
	// 2. 服务端主动关闭stream
	// 3. QUIC连接断开
	targetReq, err := http.NewRequestWithContext(streamCtx, req.Method, targetURL, req.Body)
	if err != nil {
		atomic.AddInt64(&h.stats.ErrorCount, 1)
		errorMsg := core.NewErrorMessage(h.tunnelID, requestID, 3, fmt.Sprintf("创建请求失败: %v", err))
		errorMsg.Write(stream)
		return err
	}

	// 复制请求头
	for key, values := range req.Header {
		for _, value := range values {
			targetReq.Header.Add(key, value)
		}
	}

	// 监听stream关闭事件（用于日志和错误处理）
	streamClosed := make(chan struct{}, 1)
	go func() {
		<-streamCtx.Done()
		// Stream关闭，发送取消信号
		select {
		case streamClosed <- struct{}{}:
		default:
		}
	}()

	// 🔍 性能跟踪：记录各阶段耗时
	log.Printf("⏱️  [性能] 准备发送请求: 解析耗时=%dms, 并发等待=%dms, RequestID=%s",
		parseDuration.Milliseconds(), acquireDuration.Milliseconds(), requestID[:8]+"...")

	// 发送请求
	fmt.Fprintf(os.Stderr, "🚀 [客户端] 开始HTTP请求: %s, RequestID=%s\n", time.Now().Format("15:04:05.000"), requestID[:8]+"...")
	httpStart := time.Now()
	resp, err := h.httpClient.Do(targetReq)
	httpDuration := time.Since(httpStart)
	fmt.Fprintf(os.Stderr, "✅ [客户端] HTTP响应: %s, 耗时=%.0fms, RequestID=%s\n", time.Now().Format("15:04:05.000"), httpDuration.Seconds()*1000, requestID[:8]+"...")

	if err != nil {
		// 🔥 详细错误分类和日志
		errMsg := err.Error()
		log.Printf("❌ [HTTP协议] 请求失败: %v (耗时=%.3fs), 目标=%s, RequestID=%s",
			err, httpDuration.Seconds(), targetURL, requestID[:8]+"...")
		atomic.AddInt64(&h.stats.ErrorCount, 1)

		// 检查是否是取消导致的
		select {
		case <-streamClosed:
			log.Printf("🚫 [HTTP协议] 请求已取消（浏览器断开）: RequestID=%s", requestID[:8]+"...")
			return nil // 正常取消，不返回错误
		default:
			// 真正的请求错误 - 提供更友好的错误信息
			if strings.Contains(errMsg, "connection refused") {
				log.Printf("⚠️  [HTTP协议] 目标服务未启动或端口错误: %s:%s, RequestID=%s",
					h.targetHost, h.targetPort, requestID[:8]+"...")
			} else if strings.Contains(errMsg, "timeout") {
				log.Printf("⏱️  [HTTP协议] 请求超时: 目标=%s, RequestID=%s",
					targetURL, requestID[:8]+"...")
			} else if strings.Contains(errMsg, "EOF") || strings.Contains(errMsg, "unexpected EOF") {
				log.Printf("🔌 [HTTP协议] 连接意外关闭: 目标=%s, RequestID=%s",
					targetURL, requestID[:8]+"...")
			}
		}

		errorMsg := core.NewErrorMessage(h.tunnelID, requestID, 4, fmt.Sprintf("请求失败: %v", err))
		errorMsg.Write(stream)
		return err
	}
	defer resp.Body.Close()

	atomic.AddInt64(&h.stats.SuccessCount, 1)

	log.Printf("✅ [HTTP协议] 收到响应: Status=%d, ContentLength=%d, HTTP耗时=%.3fs, RequestID=%s",
		resp.StatusCode, resp.ContentLength, httpDuration.Seconds(), requestID[:8]+"...")

	// 🚀 关键：判断是否使用流式传输
	contentLength := resp.ContentLength
	// 🎯 性能优化：降低流式传输阈值（8KB），因为分片大小已降低
	useStreaming := contentLength > int64(core.ChunkSize) || contentLength == -1 // 大于16KB或未知大小

	var sendErr error
	sendStart := time.Now()
	if useStreaming {
		log.Printf("📺 [HTTP协议] 使用流式传输: ContentLength=%d, RequestID=%s",
			contentLength, requestID[:8]+"...")
		// 🎯 使用streamCtx（无超时，只监听stream关闭）
		sendErr = h.sendResponseStreaming(&stream, requestID, resp, streamCtx, streamClosed)
	} else {
		log.Printf("📄 [HTTP协议] 使用普通传输: ContentLength=%d, RequestID=%s",
			contentLength, requestID[:8]+"...")
		sendErr = h.sendResponseNormal(&stream, requestID, resp, httpStart)
	}
	sendDuration := time.Since(sendStart)
	totalDuration := time.Since(startTime)

	stream.Close()

	// 🔍 性能总结
	log.Printf("⏱️  [性能总结] RequestID=%s: 总耗时=%.3fs (解析=%dms, HTTP请求=%.3fs, 发送响应=%.3fs)",
		requestID[:8]+"...", totalDuration.Seconds(), parseDuration.Milliseconds(),
		httpDuration.Seconds(), sendDuration.Seconds())

	return sendErr
}

// GetProtocolType 实现 ProtocolHandler 接口
func (h *Handler) GetProtocolType() core.ProtocolType {
	return core.ProtocolHTTP
}

// Close 实现 ProtocolHandler 接口
func (h *Handler) Close() error {
	h.httpClient.CloseIdleConnections()
	log.Printf("🛑 [HTTP协议] 处理器已关闭")
	return nil
}

// GetStats 获取统计信息
func (h *Handler) GetStats() core.ProtocolStats {
	return core.ProtocolStats{
		TotalRequests:  atomic.LoadInt64(&h.stats.TotalRequests),
		ActiveRequests: atomic.LoadInt64(&h.activeTasks),
		TotalBytesSent: atomic.LoadInt64(&h.stats.TotalBytesSent),
		TotalBytesRecv: atomic.LoadInt64(&h.stats.TotalBytesRecv),
		SuccessCount:   atomic.LoadInt64(&h.stats.SuccessCount),
		ErrorCount:     atomic.LoadInt64(&h.stats.ErrorCount),
	}
}

// writeChunkWithTimeout 带超时的分片写入，避免永久阻塞
func (h *Handler) writeChunkWithTimeout(stream *quic.Stream, chunk *core.ChunkMessage, ctx context.Context, streamClosed <-chan struct{}, requestID string) error {
	type writeResult struct {
		err error
	}
	writeChan := make(chan writeResult, 1)

	// 在goroutine中执行写入
	go func() {
		// 手动构建ChunkMessage格式并写入
		// 使用与服务端相同的格式
		buf := make([]byte, 0, len(chunk.Data)+100) // 预分配缓冲区

		// 写入消息类型（1字节） - 使用ResponseChunk类型
		buf = append(buf, byte(core.MessageTypeResponseChunk))

		// 写入TunnelID（36字节）
		buf = append(buf, []byte(h.tunnelID)...)

		// 写入RequestID（36字节）
		buf = append(buf, []byte(chunk.RequestID)...)

		// 写入分片索引（4字节，uint32大端序）
		chunkIndex := uint32(chunk.ChunkID)
		buf = append(buf, byte(chunkIndex>>24), byte(chunkIndex>>16), byte(chunkIndex>>8), byte(chunkIndex))

		// 写入是否最后一个分片（1字节）
		if chunk.IsLast {
			buf = append(buf, 1)
		} else {
			buf = append(buf, 0)
		}

		// 写入分片长度（4字节，uint32大端序）
		chunkLength := uint32(len(chunk.Data))
		buf = append(buf, byte(chunkLength>>24), byte(chunkLength>>16), byte(chunkLength>>8), byte(chunkLength))

		// 写入分片数据
		buf = append(buf, chunk.Data...)

		// 一次性写入所有数据
		_, err := stream.Write(buf)
		writeChan <- writeResult{err: err}
	}()

	// 等待：写入完成、context超时、stream关闭
	select {
	case result := <-writeChan:
		return result.err

	case <-streamClosed:
		log.Printf("🚫 [HTTP协议] Stream已关闭，取消写入: RequestID=%s", requestID[:8]+"...")
		stream.CancelWrite(0)
		return fmt.Errorf("stream closed")

	case <-ctx.Done():
		log.Printf("⏱️ [HTTP协议] 写入超时: RequestID=%s", requestID[:8]+"...")
		stream.CancelWrite(0)
		return fmt.Errorf("write timeout")
	}
}

// sendResponseStreaming 流式发送HTTP响应
func (h *Handler) sendResponseStreaming(stream *quic.Stream, requestID string, resp *http.Response, ctx context.Context, streamClosed <-chan struct{}) error {
	// 构建响应头
	var headerBuf bytes.Buffer
	headerBuf.WriteString(fmt.Sprintf("HTTP/1.1 %d %s\r\n", resp.StatusCode, http.StatusText(resp.StatusCode)))
	for name, values := range resp.Header {
		for _, value := range values {
			headerBuf.WriteString(fmt.Sprintf("%s: %s\r\n", name, value))
		}
	}
	headerBuf.WriteString("\r\n")

	// 🚀🚀🚀 混合模式优先策略：优先使用HTTP/2发送响应（如果可用）
	//
	// 优势：
	// 1. 无限并发Stream（vs QUIC的12个连接限制）
	// 2. 零拷贝传输（io.MultiReader直接组合头和Body）
	// 3. 成熟的HTTP/2协议栈（自动流控和多路复用）
	// 4. 降级保护（HTTP/2失败时自动降级到QUIC分片协议）
	//
	// 工作流程：
	// ┌─────────────┐   QUIC隧道   ┌──────────┐   HTTP/2   ┌──────────┐
	// │ 服务端请求  │ ────────────> │ 客户端   │ ──────────> │ 服务端响应│
	// └─────────────┘   (接收)      └──────────┘   (发送)    └──────────┘
	//
	if h.http2Pool != nil && h.http2Pool.HasConnectedClient() {
		// 选择负载最低的HTTP/2连接
		http2Client := h.http2Pool.SelectConnection()
		
		if http2Client != nil && http2Client.IsConnected() {
			activeStreams := http2Client.GetActiveStreams()
			log.Printf("🚀 [HTTP/2混合] 选择连接: 活跃Stream=%d, 请求ID=%s, 状态=%d",
				activeStreams, requestID[:8]+"...", resp.StatusCode)

			// 🔥 零拷贝传输：组合HTTP响应头和Body
			fullResponse := io.MultiReader(bytes.NewReader(headerBuf.Bytes()), resp.Body)

			// 🔥🔥🔥 在独立goroutine中发送HTTP/2响应，带超时监控
			http2ErrChan := make(chan error, 1)
			capturedClient := http2Client // 捕获变量
			go func() {
				http2ErrChan <- capturedClient.SendResponseZeroCopy(requestID, fullResponse)
			}()

			// 等待HTTP/2发送完成或超时/取消
			select {
			case err := <-http2ErrChan:
				if err != nil {
					log.Printf("⚠️ [HTTP/2混合] 发送失败，降级到QUIC: 请求ID=%s, 错误=%v", 
						requestID[:8]+"...", err)
					// 降级到QUIC分片协议（继续执行下面的代码）
				} else {
					// HTTP/2发送成功，直接返回
					log.Printf("✅ [HTTP/2混合] 响应发送完成: 请求ID=%s", requestID[:8]+"...")
					return nil
				}
			case <-ctx.Done():
				log.Printf("⏱️ [HTTP/2混合] Context已取消，降级到QUIC: 请求ID=%s",
					requestID[:8]+"...")
				// 超时/取消，降级到QUIC（继续执行下面的代码）
			case <-streamClosed:
				log.Printf("🚫 [HTTP/2混合] Stream已关闭: 请求ID=%s", requestID[:8]+"...")
				return nil
			}
		}
	}

	// 发送响应头（作为第一个分片）
	headerChunk := core.NewChunkMessage(h.tunnelID, requestID, 0, false, headerBuf.Bytes())
	if err := h.writeChunkWithTimeout(stream, headerChunk, ctx, streamClosed, requestID); err != nil {
		log.Printf("🚫 [HTTP协议] 响应头写入失败: %v, RequestID=%s", err, requestID[:8]+"...")
		return nil
	}

	log.Printf("✅ [HTTP协议] 响应头已发送: %d bytes", headerBuf.Len())

	// 🔥🔥🔥 关键修复：不使用contextReader包装resp.Body
	// 问题：contextReader会在ctx取消时返回"context canceled"错误
	// 解决：直接使用resp.Body，让它自然读取完成
	//
	// 安全性：
	// 1. 仍然在循环中检查streamClosed（浏览器断开时立即停止）
	// 2. 添加读取超时保护（避免永久阻塞）
	// 3. HTTP连接本身有自己的超时机制

	// cancelableReader := &contextReader{  // ❌ 删除
	// 	reader: resp.Body,
	// 	ctx:    ctx,
	// }

	// 🎯 使用自适应分片大小（根据带宽动态调整）
	chunkIndex := uint32(1) // 从1开始，因为0是响应头
	totalBodyBytes := int64(0)
	lastChunkSize := uint64(0)
	var buffer []byte

	// 🚀 初始化缓冲区
	adaptiveChunkSize := GetAdaptiveChunkSize()
	buffer = make([]byte, adaptiveChunkSize)
	lastChunkSize = adaptiveChunkSize

	log.Printf("🚀 [立即流式传输] ContentLength=%.2f MB, 初始分片=%d KB, RequestID=%s",
		float64(resp.ContentLength)/(1024*1024), adaptiveChunkSize/1024, requestID[:8]+"...")

	// 🔥🔥🔥 关键策略：响应头已发送，立即开始流式传输Body
	// 优势：
	// 1. 浏览器立即收到响应头，不会超时
	// 2. 边读边发，内存占用小，适合大文件
	// 3. 支持Range请求的流式传输
	// 🚀🚀🚀 流水线处理：读取和发送并行进行（高速公路模式）
	//
	// 架构说明：
	// ┌─────────┐    读取管道    ┌──────────┐    发送管道    ┌─────────┐
	// │ HTTP源  │ ──────────────> │ 缓冲队列 │ ──────────────> │ QUIC流  │
	// └─────────┘   (异步读取)   └──────────┘   (异步发送)   └─────────┘
	//                车站A                                      车站B
	//
	// 优势：
	// 1. 读取和发送解耦，互不阻塞
	// 2. 缓冲队列平滑数据流动
	// 3. 最大化利用QUIC管道

	type chunkData struct {
		index  uint32
		data   []byte
		isLast bool
	}

	// 缓冲队列：3个分片的缓冲（平衡内存和性能）
	chunkQueue := make(chan chunkData, 3)
	readError := make(chan error, 1)
	writeError := make(chan error, 1)

	// 🔥🔥🔥 创建可取消的Context,当读取错误时立即取消所有goroutine
	pipelineCtx, cancelPipeline := context.WithCancel(ctx)
	defer cancelPipeline() // 确保退出时取消所有goroutine
	
	// 🚀 车站A：异步读取goroutine
	go func() {
		defer close(chunkQueue)

		for {
			// 🔥 检查Context是否已取消
			select {
			case <-pipelineCtx.Done():
				log.Printf("🚫 [读取goroutine] Context已取消,停止读取: RequestID=%s", requestID[:8]+"...")
				return
			default:
			}
			
			// 检查是否需要调整缓冲区大小
			currentChunkSize := GetAdaptiveChunkSize()
			if currentChunkSize != lastChunkSize {
				buffer = make([]byte, currentChunkSize)
				lastChunkSize = currentChunkSize
				log.Printf("📊 [动态调整] 缓冲区大小: %d KB, RequestID=%s",
					currentChunkSize/1024, requestID[:8]+"...")
			}

			// 读取数据
			n, err := resp.Body.Read(buffer)

			if n > 0 {
				isLast := (err == io.EOF)
				totalBodyBytes += int64(n)

				// 复制数据到新缓冲区（避免并发问题）
				dataCopy := make([]byte, n)
				copy(dataCopy, buffer[:n])

				// 发送到队列
				select {
				case chunkQueue <- chunkData{
					index:  chunkIndex,
					data:   dataCopy,
					isLast: isLast,
				}:
					chunkIndex++

					if isLast {
						return // 读取完成
					}
				case <-streamClosed:
					return // Stream关闭
				case <-pipelineCtx.Done():
					return // Context取消
				}
			}

			if err != nil {
				if err == io.EOF {
					// 如果之前没有数据，发送空的最后分片
					if n == 0 && chunkIndex > 1 {
						select {
						case chunkQueue <- chunkData{
							index:  chunkIndex,
							data:   []byte{},
							isLast: true,
						}:
						case <-streamClosed:
						case <-pipelineCtx.Done():
						}
					}
					return // 读取完成
				}

				// 🔥 详细的错误分类
				errMsg := err.Error()
				if strings.Contains(errMsg, "unexpected EOF") {
					log.Printf("❌ [HTTP协议] 读取Body错误: 上游服务提前关闭连接 (RequestID=%s, 已读取=%d bytes)",
						requestID[:8]+"...", totalBodyBytes)
				} else if strings.Contains(errMsg, "timeout") {
					log.Printf("❌ [HTTP协议] 读取Body超时 (RequestID=%s, 已读取=%d bytes)",
						requestID[:8]+"...", totalBodyBytes)
				} else {
					log.Printf("❌ [HTTP协议] 读取Body错误: %v (RequestID=%s, 已读取=%d bytes)",
						err, requestID[:8]+"...", totalBodyBytes)
				}
				
				// 🔥🔥🔥 关键修复：取消Context,立即停止所有goroutine
				cancelPipeline()
				
				// 真正的错误
				readError <- err
				return
			}
		}
	}()

	// 🚀 车站B：异步发送goroutine
	go func() {
		sentCount := uint32(0)

		for {
			select {
			case chunkInfo, ok := <-chunkQueue:
				if !ok {
					// 队列关闭，读取完成
					log.Printf("✅ [写入goroutine] 队列已关闭,停止发送: RequestID=%s", requestID[:8]+"...")
					return
				}

				// 创建分片消息
				chunk := core.NewChunkMessage(h.tunnelID, requestID, uint64(chunkInfo.index), chunkInfo.isLast, chunkInfo.data)

				// 发送分片并记录性能
				writeStart := time.Now()
				writeErr := h.writeChunkWithTimeout(stream, chunk, pipelineCtx, streamClosed, requestID)
				writeDuration := time.Since(writeStart)

				if writeErr != nil {
					log.Printf("🚫 [HTTP协议] Stream写入失败: err=%v, 耗时=%.2fs, RequestID=%s",
						writeErr, writeDuration.Seconds(), requestID[:8]+"...")
					
					// 🔥🔥🔥 取消Context,停止读取goroutine
					cancelPipeline()
					
					writeError <- writeErr
					return
				}

				// 记录性能统计
				globalBandwidthAdapter.RecordSuccess(int64(len(chunkInfo.data)), writeDuration)

				sentCount++

				// 每20个分片打印进度
				if sentCount%20 == 0 {
					chunkSize, bandwidth := globalBandwidthAdapter.GetStats()
					log.Printf("📤 [HTTP协议] 已发送 %d 个分片, Body: %.2f MB (分片: %d KB, 估算带宽: %.2f KB/s)",
						sentCount, float64(totalBodyBytes)/(1024*1024), chunkSize/1024, bandwidth)
				}

				if chunkInfo.isLast {
					log.Printf("✅ [HTTP协议] 流式响应完成: %d 分片, %.2f MB",
						sentCount, float64(totalBodyBytes)/(1024*1024))
					return
				}

			case <-streamClosed:
				log.Printf("🚫 [写入goroutine] Stream已关闭,停止发送: RequestID=%s", requestID[:8]+"...")
				
				// 🔥 取消Context
				cancelPipeline()
				return
				
			case <-pipelineCtx.Done():
				log.Printf("🚫 [写入goroutine] Context已取消,停止发送: RequestID=%s", requestID[:8]+"...")
				return
			}
		}
	}()

	// 等待读取或发送完成
	select {
	case err := <-readError:
		log.Printf("❌ [HTTP协议] 读取Body错误: %v (RequestID=%s)", err, requestID[:8]+"...")
		return nil
	case err := <-writeError:
		log.Printf("❌ [HTTP协议] 写入错误: %v (RequestID=%s)", err, requestID[:8]+"...")
		return nil
	case <-streamClosed:
		log.Printf("🚫 [HTTP协议] Stream已关闭: RequestID=%s", requestID[:8]+"...")
		return nil
	case <-ctx.Done():
		log.Printf("⏱️  [HTTP协议] Context已取消: RequestID=%s", requestID[:8]+"...")
		return nil
	}
}

// sendResponseNormal 普通方式发送HTTP响应（小文件）
func (h *Handler) sendResponseNormal(streamPtr *quic.Stream, requestID string, resp *http.Response, startTime time.Time) error {
	stream := *streamPtr
	var respBuf bytes.Buffer

	// 写入状态行
	respBuf.WriteString(fmt.Sprintf("HTTP/1.1 %d %s\r\n", resp.StatusCode, http.StatusText(resp.StatusCode)))

	// 写入响应头
	for name, values := range resp.Header {
		for _, value := range values {
			respBuf.WriteString(fmt.Sprintf("%s: %s\r\n", name, value))
		}
	}
	respBuf.WriteString("\r\n")

	// 写入Body
	bodySize, err := io.Copy(&respBuf, resp.Body)
	if err != nil {
		log.Printf("❌ [HTTP协议] 读取响应Body失败: %v", err)
		errorMsg := core.NewErrorMessage(h.tunnelID, requestID, 5, fmt.Sprintf("读取响应失败: %v", err))
		errorMsg.Write(stream)
		return err
	}

	// 发送响应消息
	responseMsg := &core.TunnelMessage{
		Version:      core.ProtocolVersion,
		MessageType:  core.MessageTypeResponse,
		ProtocolType: core.ProtocolHTTP,
		TunnelID:     h.tunnelID,
		RequestID:    requestID,
		Body:         respBuf.Bytes(),
		BodyLength:   uint64(respBuf.Len()),
	}

	if err := responseMsg.Write(stream); err != nil {
		log.Printf("❌ [HTTP协议] 发送响应失败: %v", err)
		return err
	}

	duration := time.Since(startTime)
	log.Printf("✅ [HTTP协议] 响应发送完成: RequestID=%s, Status=%d, Body=%d bytes, 耗时=%v",
		requestID[:8]+"...", resp.StatusCode, bodySize, duration)

	return nil
}
